package service

import (
	"fmt"
	"time"

	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/contract"
	ecom "hundsun.com/hsl/hschain/executor/common"
	pb "hundsun.com/hsl/hschain/protos/execute"
	ptypes "hundsun.com/hsl/hschain/protos/types"
	extledger "hundsun.com/hsl/hschain/store/extend/ledger"
	"hundsun.com/hsl/hschain/store/mq"
)

// work模式处理msg
func (w *WorkerService) procWorkMqMsg() {
	for msg := range w.workClient.Recv() {
		w.log.Tracef("work recv msg ty %s", types.GetEventName(int(msg.Ty)))
		msgtype := msg.Ty
		switch msgtype {
		case types.EventQueryContract:
			go mq.ProcessMsg(msg, w.procExecQuery, w.workClient, w.log)
		case types.EventExecTxGroup:
			go mq.ProcessMsg(msg, w.procExecTxGroups, w.workClient, w.log)
		default:
		}
	}
}

func (w *WorkerService) procExecQuery(msg *mq.Message) {
	log := w.log
	client := w.workClient
	var err error
	defer func() {
		if r := recover(); r != nil { // 不能panic掉
			err = fmt.Errorf("panic error %v ", r)
		}
		if err != nil {
			log.Errorf("procExecQuery %v ", err)
			msg.Reply(w.workClient.NewMessage("", "", types.EventReplyQueryContract, &ptypes.Reply{IsOk: false, Msg: []byte(err.Error())}))
			return
		}
	}()

	ldIns, err := extledger.NewLedger(w.ledgerDB)
	if err != nil {
		log.Errorf("procExecQuery NewLedger err %s", err.Error())
		return
	}
	blkNum, err := ldIns.GetMaxExecutedBlockNumber()
	if err != nil {
		log.Errorf("execQuery get blkNum %d from ledger error %v", blkNum, err)
		return
	}
	blk, err := ldIns.GetBlock(blkNum)
	if err != nil {
		log.Errorf("execQuery get block %d from ledger error %v", blkNum, err)
		return
	}
	blkExecRes, err := ldIns.GetBlockExecResult(blkNum)
	if err != nil {
		log.Errorf("execQuery get GetBlockExecResult %d from ledger error %v ", blkNum, err)
		return
	}
	ctx := ecom.NewExecContent(
		blk.GetHeader(),
		blkExecRes.GetStateHash(),
		w.ledgerDB,
		ecom.NewStateHandle(nil, nil, w.stateDB),
	)

	data := msg.Data.(*types.QueryExecutor)
	driver, err := contract.LoadDriver(data.Driver, blkNum)
	if err != nil {
		log.Errorf("execQuery load contract driver fail, exec type is %s", data.Driver)
		return
	}
	defer contract.AddDriverHandle(data.Driver, driver)
	rep, err := driver.Query(ctx, data.FuncName, data.Param)
	if err != nil {
		log.Errorf("execQuery query fail error %v, funcName is：%s", err, data.FuncName)
		return
	}
	value, _ := proto.Marshal(rep)
	// replay
	msg.Reply(client.NewMessage("", "", types.EventReplyQueryContract, &ptypes.Reply{IsOk: true, Msg: value}))
}

func (w *WorkerService) procExecTxGroups(msg *mq.Message) {
	log := w.log
	var err error
	defer func() {
		if r := recover(); r != nil { // 不能panic掉
			err = fmt.Errorf("panic error %v ", r)
		}
		if err != nil {
			log.Errorf("procExecTxGroups %v ", err)
			msg.Reply(w.workClient.NewMessage("", "", types.EventReplyExecTxGroup, &ptypes.Reply{IsOk: false, Msg: []byte(err.Error())}))
			return
		}
	}()

	if msg.Data == nil {
		err = fmt.Errorf("msg data is nil")
		log.Errorf("procExecTxGroups %v", err)
		return
	}
	value := msg.Data.([]byte)
	if value == nil {
		err = fmt.Errorf("msg data in is nil")
		log.Errorf("procExecTxGroups %v", err)
		return
	}
	in := &pb.TxGroupsRequest{}
	err = proto.Unmarshal(value, in)
	if err != nil {
		err = fmt.Errorf("msg data Unmarshal fail")
		log.Errorf("procExecTxGroups %v", err)
		return
	}

	groups := in.Groups
	blockNumber := in.BlockNumber
	mqLog := log.WithField(logging.LogFieldBlockNumber, blockNumber).WithField(logging.LogFieldWorkerID, w.conf.WorkerID)

	mqLog.Debugf("Start execute transaction groups size = [%d]", len(groups))

	workerID := in.WorkerId
	currentBlockHeader := in.CurrentBlockHeader
	preBlockExecResults := in.PreStateDatas
	startTime := time.Now()
	newGroups, hasNewAddr, txCount, err := w.ParallelExecute(groups, currentBlockHeader, preBlockExecResults, w.Execute)
	if err != nil {
		mqLog.Errorf("ParallelExecute groups failed,error is %s", err.Error())
		return
	}

	mqLog.Infof("ParallelExecute groupsTxs %d groups %d cost time %v ", txCount, len(groups), time.Since(startTime))

	rep := &pb.TxGroupsReply{
		BlockNumber: blockNumber,
		WorkerId:    workerID,
		HasNewAddr:  hasNewAddr,
		Groups:      newGroups,
	}
	value, err = proto.Marshal(rep)
	if err != nil {
		mqLog.Errorf("ParallelExecute groups marshal fail error %s", err.Error())
		return
	}
	// replay
	msg.Reply(w.workClient.NewMessage("", "", types.EventReplyExecTxGroup, &ptypes.Reply{IsOk: true, Msg: value}))
}
